package org.myfram.flink.flinkonjar.domain.process.kafkaJoinHbase;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.functions.MapFunction;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.KafkaSourceUserEvent;

public class KafkaUserFieldProcessFunction implements MapFunction<String, KafkaSourceUserEvent> {

    private static final ObjectMapper mapper = new ObjectMapper();

    @Override
    public KafkaSourceUserEvent map(String value) throws Exception {
        JsonNode node = mapper.readTree(value);
        String id = node.get("id").asText();
        String name = node.get("name").asText();
        int age = node.has("age") ? node.get("age").asInt() : 0;
        return new KafkaSourceUserEvent(id, name);
    }
}
